package com.bungarus.busi;

import com.bungarus.Busi;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.*;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * The remedy lives at Busi node must send the non-ack message to the correct access node according to
 * the value of 'from' property in message.
 *
 * Created by tang on 2018/12/23.
 */
public class BusiRemedy extends Remedy {
    private Map<String, List<InnerClient>> accessClients;
    private static boolean started = false;

    public synchronized static void start(BungarusServer server0) throws Exception {
        if(!started) {
            started = true;
            new BusiRemedy().run(server0);
        }
    }

    @Override
    protected void run(BungarusServer server) throws Exception {
        accessClients = ((Busi)server).getInnerClientList();
        executorGroup.scheduleAtFixedRate(() -> {
            DischargeQueue queue = Discharger.getQueueInstance();
            BusiRemedy.this.retry0(queue);

            logger.info("Remedy is running...");
        },DELAY_TIME ,PERIOD , TimeUnit.SECONDS);

        logger.info("Remedy is running...");
    }

    private void retry0(DischargeQueue queue) {
        logger.debug("Size of discharge queue: " + queue.size());
        if(queue.size() <= 0 || accessClients.isEmpty()){
            return;
        }
        ConcurrentHashMap.KeySetView<Long, MessageProtobuf.Message> keys = queue.keys();
        for(Long l: keys){
            MessageProtobuf.Message noAckMsg = queue.getFromQueue(l);
            if((System.currentTimeMillis() - noAckMsg.getHeader().getSentTime()) >= ACK_TIMEOUT) {
                List<InnerClient> clients = accessClients.get(Busi.CLUSTER_TAG_ACCESS);

                Set sessions = SessionManagement.getInstanct().retrieve(noAckMsg.getHeader().getFrom(),
                        String.valueOf(noAckMsg.getHeader().getTenantID()));
                sessions.stream().forEach(session -> {
                    String[] values = StringUtils.delimitedListToStringArray(session.toString(), SessionManagement.VALUE_SEPERATOR);
                    String reciptId = values[0];
                    String sessionId = values[1];
                    String ipport = values[2];
                    for(InnerClient accessClient: clients) {
                        if(ipport.equals(accessClient.getSrvIp() + ":" + accessClient.getSrvPort())) {
                            try {
                                if(MessageStatus.REMEDY == noAckMsg.getHeader().getReturnCode()) {
                                    accessClient.saveTryFailedMessage(noAckMsg);
                                }
                                else {
                                    accessClient.sendMessage(MessageBuilder.buildRemedyMessage(noAckMsg), Busi.CLUSTER_TAG_ACCESS);
                                }
                                queue.removeFromQueue(l);
                            }
                            catch (BungarusClientException e) {
                                e.printStackTrace();
                            }
                            break;
                        }
                    }
                });
            }
        }
    }
}
